package com.kafka.producer.producer;

import jakarta.annotation.Resource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

import static com.kafka.producer.config.RedisConfig.*;

@Component
public class Producer {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String key, String str) {

        try {
            // 非阻塞监听消息回调
            CompletableFuture<SendResult<String, String>> res = kafkaTemplate.send(TOPIC, 2, System.currentTimeMillis(), key, str);
            res.thenAccept((result) -> {
                if(result.getRecordMetadata() != null){
                    System.out.println("发送成功" + result.getRecordMetadata().toString());
                }
                System.out.println("record：" + result.getProducerRecord());
            }).exceptionally((t) -> {
                // 失败处理
                // 手动异常重发
                // 放入到 redis 中，待会再重试发送
                stringRedisTemplate.opsForList().rightPush(MSG_LIST+TOPIC, key+":"+str+":"+0);
                System.out.println("发送失败异常，放入到 redis 中，待会再重试发送");
                return null;
            });
        } catch (Exception e) {
            // 手动异常重发
            // 放入到 redis 中，待会再重试发送
            stringRedisTemplate.opsForList().rightPush(MSG_LIST+TOPIC, key+":"+str+":"+0);
            System.out.println("发送失败异常，放入到 redis 中，待会再重试发送");
            throw new RuntimeException(e);
        }
    }

    public void sendMessage(String key, String str, int retryCount) {

        try {
            // 非阻塞监听消息回调
            CompletableFuture<SendResult<String, String>> res = kafkaTemplate.send(TOPIC, 2, System.currentTimeMillis(), key, str);
            res.thenAccept((result) -> {
                if(result.getRecordMetadata() != null){
                    System.out.println("发送成功" + result.getRecordMetadata().toString());
                    // 移除出redis
                    stringRedisTemplate.opsForList().leftPop(MSG_LIST+TOPIC);
                }
                System.out.println("record：" + result.getProducerRecord());
            });
        } catch (Exception e) {
            // 手动异常重发
            // 判断重试次数是否达到上限（3次）
            if(retryCount > RETRIES) {
                // 超过次数，则不重试
                // 移除出redis队列中
                stringRedisTemplate.opsForList().leftPop(MSG_LIST+TOPIC);
                System.out.println("发送失败异常，超过次数，不再重试");
            } else {
                // 没有超出次数，则重新放入到 redis 中，待会再重试发送
                // 重试次数+1
                int count = retryCount + 1;
                // 更新重试次数
                stringRedisTemplate.opsForList().set(MSG_LIST+TOPIC, 0,  key+":"+str+":"+count);
                System.out.println("再次发送失败，重新放入到 redis 中，待会再重试发送");
            }
            throw new RuntimeException(e);
        }
    }
}
